package ac.uk.qmul.gemom.scenario;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.text.DateFormat;
import java.text.SimpleDateFormat;

import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;
 

import java.math.BigDecimal;

import com.qsphere.boundary.gemom.GEMOMClientData;
import com.qsphere.boundary.gemom.GEMOMData;
import com.qsphere.boundary.gemom.GEMOMBrokerAdapter;
import com.qsphere.boundary.gemom.GEMOMListener;
import com.qsphere.boundary.gemom.GEMOMMessageType;
import com.qsphere.boundary.gemom.GEMOMParameters;
import com.qsphere.boundary.gemom.impl.GEMOMBrokerAdapterImpl;

import inl.resilienceManager.qmul.gemom.iniFrame;


/*
 * A class used for counting message rates 
 * 
 * COPYRIGHT DISCLAIMER
 * Synopsis: All the source codes presented in this document are intellectual
 * properties of Queen Mary, University of London.
 * 
 * @year 2010
 * @author Peng Jiang, John Bigham, Jiayi Wu Jinfu Wang------------QMUL
 * @version 1.0 
 */

public class GEMOMCounter extends Thread{
	
	private Subscription subscription;

	private boolean initialised;
	private boolean non_stop;
	
	private String model = "";
	
	private int timeUnit;    // time unit of counters in milli-seconds
	
	private Date date;
	
	private double time;
	
	private double stdAlphaEWMA;
	
	private Hashtable<String, EWMACounter> rawEwmaCounters;
	
	private Hashtable<String, EWMACounter> ewmaCounters;
	
	private ScenarioFileParser scenarioFile;
	 
	private String clientName;
		
	private GEMOMBrokerAdapter  brokerAdapter;
	
	protected final double default_alpha = 0.2;
	
	private String publisherStatus;
	
	private boolean loggingStarted;
	
	private int loggingCounter;
	
	private long loggingStartTime = 0;
	
	private iniFrame masterFrame;
	
	ArrayList<String> loggedLine; 
	 
	
	// level 0: total message rate for all namespaces, 
	//       1: total message rate for all namespaces, messege rate for individual namespace, message rate for individual topics 
	protected int counterPublishLevel;

	public GEMOMCounter(iniFrame frame, GEMOMBrokerAdapter adapter, ScenarioFileParser fp, String cname, Subscription sub, int publishLevel){
		this.masterFrame = frame;
		
		this.subscription = sub;
		this.brokerAdapter = adapter; 
		this.clientName = cname;
		this.scenarioFile = fp;
		this.counterPublishLevel = publishLevel;
		this.publisherStatus = "";
		loggingCounter = 0;
		initialised = false;
		// unless having received a signal from the source, the logging will not start
		loggingStarted = false;
		date = new Date();
		
	}
	
	public boolean isInitialised(){
		return initialised;
	}
	
	public void setPublishLevel(int publishLevel){
		this.counterPublishLevel = publishLevel;
	}
	
	public String getModelName(){
		return model;
	}
		
	public void initEWMA(double alpha){
		
		if(alpha < 0 || alpha >1)
			stdAlphaEWMA = this.default_alpha;
		else
		    stdAlphaEWMA = alpha;
		
		rawEwmaCounters = null;
		
		rawEwmaCounters = new Hashtable<String, EWMACounter>();
		
		ewmaCounters = null;
		
		ewmaCounters = new Hashtable<String, EWMACounter>();
		
		Enumeration nss  = subscription.getAllNamespace().keys();
		//int size = 0;
		while(nss.hasMoreElements()){
			String ns = nss.nextElement().toString();
			if (ns != null || !ns.equalsIgnoreCase("")) {
				//size++;
				
				rawEwmaCounters.put(ns, new EWMACounter(ns, alpha));
				ewmaCounters.put(ns, new EWMACounter(ns, alpha));

				Vector<String> topics  = subscription.getNamespace(ns);
				for(int i = 0; i < topics.size(); i ++){
					String name = topics.get(i);
					name = name+"@/"+ns;
					rawEwmaCounters.put(name, new EWMACounter(name, alpha));
				}
				//size += subscription.getNamespaceSize(ns);
			}       
		}	
        model = "EWMA";
        this.timeUnit = 1000;
        
        time = date.getTime();
        
	    non_stop = true;
	    
	    if(scenarioFile != null && subscription != null && brokerAdapter != null
	    		&& clientName != null && !clientName.equalsIgnoreCase("")){
	    	
	        initialised = true;
	        
	        // empty the namespace, get it ready for new publish
	        brokerAdapter.deleteNamespace(null, scenarioFile.counter_namespace, null, null);
	    } 
	    
	}
	
	
	public void end(){
		this.non_stop = false;
		this.initialised = false;
	}
	
	public void run(){
		if(initialised){ 
			  
			resetLogger();
			
			while(non_stop){
		
				try{
				    this.sleep((int)timeUnit);
				    if(model.equalsIgnoreCase("EWMA"))
				        updateAllEWMA();
				    
				    /*
				    if(brokerAdapter != null)
				    System.out.println("broker = "+brokerAdapter);
				    if(subscription != null)
				    System.out.println("sub = "+subscription);
				    if(scenarioFile != null)
				    System.out.println("fp = "+scenarioFile);
				    */ 
				    // log the data, only starts when having received a signal
				    // logging function commented, logging is done by a client which
				    // receives data from counter namespace
				    log();
					
				    publish();
				      
				}
				catch(Exception e){
					System.err.println(e);
				}
			}
		}
		
	}
	
	private double fitValue(double value){

		// round the observed message rate and keep maximum 2 digits to the right of 
		// the decimal point
		 
	    java.text.NumberFormat formate = java.text.NumberFormat.getNumberInstance();  
	    formate.setMaximumFractionDigits(3);
	    String m = formate.format(value); 
	    double d = Double.parseDouble(m);
		//BigDecimal bd = new BigDecimal(Double.toString(value));  
        //bd = bd.setScale(3,BigDecimal.ROUND_HALF_EVEN);  
        //double d = bd.doubleValue();   
		return d;
	}
	
	private void publish(){ 
		GEMOMData data = new GEMOMData();
		String sysName = scenarioFile.counter_namespace;
	
    	Enumeration counterkeys  = rawEwmaCounters.keys();
		//int size = 0;
    	//System.out.println("size ="+rawEwmaCounters.size());
		while(counterkeys.hasMoreElements()){
			String name = counterkeys.nextElement().toString();
			if (name != null || !name.equalsIgnoreCase("")) {
				double value = fitValue(rawEwmaCounters.get(name).getEWMA());
				// view message rate at all levels
				if(counterPublishLevel == 1)
				    data.put(name, ""+value);
				// view message rate at topic level only
				if(counterPublishLevel == 2 && name.contains("@"))
				    data.put(name, ""+value);
			}       
		}	
		//System.out.println("size 2 ="+ewmaCounters.size());
		// updates for namespace message rate
		counterkeys  = ewmaCounters.keys();
		while(counterkeys.hasMoreElements()){
			String name = counterkeys.nextElement().toString();
			if (name != null || !name.equalsIgnoreCase("")) {
				double value = fitValue(ewmaCounters.get(name).getEWMA());
				if(counterPublishLevel == 1)
				    data.put(name, ""+value);
				// view message rate at namespace level only
				if(counterPublishLevel == 3)
				    data.put(name, ""+value);
			}       
		}	
		
		String name = "Total msg rate";
		double value = fitValue(getTotalMsgUpdateRateEWMA());
		data.put(name, ""+value);
		
		
		this.brokerAdapter.publish(null, sysName, "Msg rate", data,
				null, null);
		
	}
	
	/**
	 * Publish very specific messages, used for control/coordinating clients mainly
	 */
	protected void publish(String namespace, String topic, String name, String value) {
		
		if (topic == null || topic.equalsIgnoreCase("")) {
			return;
		}
		GEMOMData data = new GEMOMData();
		data.put(name, value);
   
		this.brokerAdapter.publish(null, namespace, topic, data,
				null, null); 
	}
	
	private void logIndividual(int i, int imax){
		String[] nsnames={"A","B","C"};
		
//		double value = fitValue(ewmaCounters.get("A").getEWMA());
		if (i<=imax){
			String s=i+"";
			for (int j=0; j<nsnames.length;j++){
				double value = fitValue(ewmaCounters.get(nsnames[j]).getEWMA());
				s=s+"\t"+value;
			}
//			output o
			loggedLine.add(s);
		}
		else if (i==imax+1){
			loggedLine.add(0, "% "+loggedLine.size());
			loggedLine.add(1, "% "+(nsnames.length+1));
			loggedLine.add(2, "% 9\t1\t1\t1");
			loggedLine.add(3, "% Key\tA\tB\tC");

			try {
				TextFile file= new TextFile(loggedLine);
				file.write("c:\\rate.lrn");
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		else return;
	}
	
	// Log the message rate counted, note this function only records message published for
	// each namespace. The message rate for each namespace is the sum of message rate of
	// all topics under the namespace, which is effectively the EWMA values of all counter
	// elements recorded in the HashTable ewmaCounter. Note when logging, the values for
	// control and counter namespaces are not recorded. 
	private void log(){ 
		
		// Auto-detect and log first line
		if (publisherStatus.equalsIgnoreCase(DemoResource.PUBLISHER_STATUS_SENDING) &&
				!loggingStarted){
			loggingStarted = true;
			String s=loggingCounter+""; 
			Enumeration counterkeys  = ewmaCounters.keys();
			//int size = 0;
			while(counterkeys.hasMoreElements()){
				String name = counterkeys.nextElement().toString();
				if (name != null || !name.equalsIgnoreCase("")) {
					//exclude data on control and rate namespaces 
					if(!name.equalsIgnoreCase(DemoResource.RATE_NAMESPACE) &&
							!name.equalsIgnoreCase(DemoResource.CONTROL_NAMESPACE)){ 
					    s=s+"\t"+name;
					}
				}       
			}	 
				
			loggedLine.add(s);
			loggingCounter++;

		}
		if (publisherStatus.equalsIgnoreCase(DemoResource.PUBLISHER_STATUS_SENDING) &&
				loggingStarted){ 
			String s=loggingCounter+""; 
			Enumeration counterkeys  = ewmaCounters.keys();
			//int size = 0;
			while(counterkeys.hasMoreElements()){
				String name = counterkeys.nextElement().toString();
				if (name != null || !name.equalsIgnoreCase("")) {
					//exclude data on control and rate namespaces 
					if(!name.equalsIgnoreCase(DemoResource.RATE_NAMESPACE) &&
							!name.equalsIgnoreCase(DemoResource.CONTROL_NAMESPACE)){
					    double value = fitValue(getNsMsgUpdateRateEWMA(name));
					
					    s=s+"\t"+value;
					}
				}       
			}	 
				
			loggedLine.add(s);
			loggingCounter++;
			
		}
		else if (publisherStatus.equalsIgnoreCase(
				DemoResource.PUBLISHER_STATUS_FINISHED) && loggingStarted){  

			// reset, or else the logging will continue
			loggingStarted = false; 
		}
		else return;
	}
	
	
	public double[][] getLoggedData(){ 
		
		  int length = loggedLine.size()-1;
	        // note the first column is reserved for index nunmbers, and is not a message rate
	        // that needs to be recorded
	        int number = (loggedLine.get(0).split("\t")).length-1;
	        
	        double[][] series = new double[length][number];
	         
	     // avoid the first row which is the ID of namespace
	        for(int i = 1; i < loggedLine.size(); i++){
	        	String[] row = loggedLine.get(i).split("\t");
	   
	        	// avoid the first column which is the index of data
	        	for(int j = 1; j < row.length; j++){
	        		series[i-1][j-1] = Double.parseDouble(row[j]);
	        	}
	        } 
			return series;
		  
	}

	
    public String[] getLoggedNamespace(){ 
	
    	
        int number = (loggedLine.get(0).split("\t")).length-1; 
	     // retireve column names = namespace ID
	        String[] nss = new String[number];
	        String[] nsNameRow = loggedLine.get(0).split("\t");
	        
	       	for(int i = 1; i < number+1; i++){
	    		nss[i-1] = nsNameRow[i];

	       	}
			
	       	return nss;
 
	}
	
	public boolean logDataIntoFile(File file){
		
		try {
			/*
			TextFile file= new TextFile(loggedLine);
			System.out.println("logging...");
			file.write("c:\\rate.log");
			*/
			
		    PrintWriter out = new PrintWriter(
		    	      new BufferedWriter(new FileWriter(file)));
		     
    	    for(int i = 0; i < loggedLine.size(); i++)
    	      out.println(loggedLine.get(i));
    	    out.close();
    	    
    	    return true;
			 
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return false;
		}
		
	}
	
	private void resetLogger(){
		loggedLine= new ArrayList<String>(); 
		loggingCounter=0;
		loggingStarted = false;
	}
	
	// Not used
	//
	// check if data is still valid. If the time has passed for too long then it
	// is probably suggesting there is something wrong with the training series
	// and that data should not be used. The default maximum idle period for when
	// conseuctive data has not been received and the series is still valid depends
	// on the argument of the method (in the unit of seconds)
	private boolean checkDataStillValid(int maxDifference){
		if(maxDifference*1000 < (System.currentTimeMillis() - loggingStartTime))
			return false;
		else 
			return true; 
	} 
		
	// Not used
	// mark starting time at which the publisher signify the start of a training
	// serie (data has not been actually sent yet)
	private void markLoggingTime(){ 
	    loggingStartTime = System.currentTimeMillis(); 
	}
	 
	
	// updates per second for a specific topic
	public double getTopicMsgUpdateRateEWMA(String namespace, String topic){
		String key = topic+"@/"+namespace;
		if(rawEwmaCounters.containsKey(key))
			return rawEwmaCounters.get(key).getEWMA();
		else
	        return -1;
	}
	
	// updates per second for a specific namespace
	public double getNsUpdateRateEWMA(String namespace){
		String key = namespace;
		if(rawEwmaCounters.containsKey(key))
			return rawEwmaCounters.get(key).getEWMA();
		else
	        return -1;
	}
	
	// updates per second for all topics under a namespace
	public double getNsMsgUpdateRateEWMA(String namespace){
		String key = namespace;
		if(ewmaCounters.containsKey(key))
			return ewmaCounters.get(key).getEWMA();
		else
	        return -1;
	}
	
	// updates per second for all topics under all namespaces
	public double getTotalMsgUpdateRateEWMA(){
		double total = 0;
		Enumeration counterkeys  = ewmaCounters.keys();
		//int size = 0;
		while(counterkeys.hasMoreElements()){
			String name = counterkeys.nextElement().toString();
			if (name != null || !name.equalsIgnoreCase("")) {
				total += ewmaCounters.get(name).getEWMA();
			}       
		}	
		 
	    return total;
	}
	
	public void receiveCommand(String status){
		this.publisherStatus = status;
		//System.out.println("here is " + status);
	}
	
	// make sure this is called everytime there is a new updates, called
	// by a GEMOM client
	public void updateToCounter(String namespace, String topic, GEMOMData data){
		
		if(initialised && model.equalsIgnoreCase("EWMA")){
			String key = "";
			if(topic == null || topic.equalsIgnoreCase(""))
				key = namespace;
				
			else{
				key = topic+"@/"+namespace;
				
				// udpate for message rates in a namespace
				if(ewmaCounters.containsKey(namespace))
					ewmaCounters.get(namespace).updateObservation(1.0);

				else{
					EWMACounter counter2 = new EWMACounter(namespace, stdAlphaEWMA);
					counter2.updateObservation(1.0);
					ewmaCounters.put(namespace, counter2);
				}
			}
			

			if(rawEwmaCounters.containsKey(key)){
				// update the observations by one - one more update in this time slot
				rawEwmaCounters.get(key).updateObservation(1.0);
				
			}
			// add new counters
			else{
				EWMACounter counter = new EWMACounter(key, stdAlphaEWMA);
				counter.updateObservation(1.0);
				rawEwmaCounters.put(key, counter);
				
			}	
		}
	}
	
	// make sure this is called everytime there is a new update, called
	// by a GEMOM client
	public void addToCounter(String namespace, String topic){
		
		if(initialised && model.equalsIgnoreCase("EWMA")){
			String key = "";
			if(topic == null || topic.equalsIgnoreCase(""))
				key = namespace;
				
			else{
				key = topic+"@/"+namespace;

			}
			
			// udpate for message rates in a namespace
			if(!ewmaCounters.containsKey(namespace)) {
				EWMACounter counter2 = new EWMACounter(namespace, stdAlphaEWMA);
				ewmaCounters.put(namespace, counter2);
			}

			if(!rawEwmaCounters.containsKey(key)){ 
				EWMACounter counter = new EWMACounter(key, stdAlphaEWMA); 
				rawEwmaCounters.put(key, counter);
				
			}
			
		}
		
	}
	
	public void removeNamespaceFromCounter(String namespace){
		
		if(initialised && model.equalsIgnoreCase("EWMA")){

			// udpate for message rates in a namespace
			if(ewmaCounters.containsKey(namespace)) {
				ewmaCounters.remove(namespace);
			}
			 
			Enumeration counterkeys  = rawEwmaCounters.keys();
			//int size = 0;
			while(counterkeys.hasMoreElements()){
				String name = counterkeys.nextElement().toString();
				if (name != null && !name.equalsIgnoreCase("")
						&& name.contains(namespace)) {
					rawEwmaCounters.remove(name);
				}       
			}	
			
		}
		
	}
	
	public void removeTopicFromCounter(String namespace, String topic){
		
		if(initialised && model.equalsIgnoreCase("EWMA")){

			
			if(topic == null || topic.equalsIgnoreCase(""))
				return;
			
			String key = topic+"@/"+namespace;
			
			/*
			
			Enumeration counterkeys  = rawEwmaCounters.keys();
			int count = 0;
			boolean hasNamespace = false;
			//int size = 0;
			while(counterkeys.hasMoreElements()){
				String name = counterkeys.nextElement().toString();
				if (name != null && !name.equalsIgnoreCase("")
						&& name.contains(namespace)) {
					if(name.equalsIgnoreCase(namespace))
						hasNamespace = true;
					count++;
				}       
			}	
			
			// need to clear statistics for the whole namespace
			if(count == 2 && hasNamespace){
				rawEwmaCounters.remove(namespace);
				ewmaCounters.remove(namespace);
			}
			*/
			
			// remove update rates for a topic
			if(rawEwmaCounters.containsKey(key)) {
				rawEwmaCounters.remove(key);
			} 
		} 
	}
	 
	// make sure this is called everytime there is a new updates
	private void updateAllEWMA(){

    	Enumeration counterkeys  = rawEwmaCounters.keys();
		//int size = 0;
		while(counterkeys.hasMoreElements()){
			String name = counterkeys.nextElement().toString();
			if (name != null || !name.equalsIgnoreCase("")) {
				EWMACounter counter = rawEwmaCounters.get(name);
				counter.updateEWMA();
			}       
		}	
		
		counterkeys  = ewmaCounters.keys();
		//int size = 0;
		while(counterkeys.hasMoreElements()){
			String name = counterkeys.nextElement().toString();
			if (name != null || !name.equalsIgnoreCase("")) {
				EWMACounter counter = ewmaCounters.get(name);
				counter.updateEWMA();

			}       
		}	
	}
}

class EWMACounter{
	

	private double EWMA;
	
	private double observation;

	private double alphaEWMA;	
	
	private String name;
	
	private boolean firstObservation;
	
	public EWMACounter(String name, double alpha){
		this.name = name;
		firstObservation = true;
		observation = 0;
		alphaEWMA = alpha;
		
	}
	
	
	public void setAlphaEWMA(double alpha){
		this.alphaEWMA = alpha;
	}
	
	// update the current count with newly observed value
	public void updateObservation(double obser){
     	observation += obser;
	}
	
	// update the EWMA value, if weight alpha is 1 then it's effectively raw data not EWMA
	public void updateEWMA(){
		if(firstObservation){
			EWMA = observation;
			firstObservation = false;
			observation = 0;
		}
		else{
		    EWMA = (1-alphaEWMA)*EWMA+ alphaEWMA*this.observation;
		    // reset observation to zero, so it can go through the whole updated cycle within the next time unit
		    observation = 0;
		}
	}
	
	// update the EWMA value
	public double getEWMA(){
		return EWMA;
		
	}
	
	
}
